package cn.com.otpexpress.job.filter.consumer;

import cn.com.otpexpress.job.filter.service.OrderFilterService;
import lombok.extern.slf4j.Slf4j;
import net.ytoframework.kernel.common.util.StringUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.UUID;

/**
 * Created by dongxing
 */
@Slf4j
@Component
public class YTOOrderConsumer implements InitializingBean{
    @Value("${yto.order.nameServer}")
    private String nameServer;
    @Value("${yto.order.online.topic}")
    private String onlineTopic;
    @Value("${yto.order.offline.topic}")
    private String offlineTopic;
    @Autowired
    private OrderFilterService orderFilterService;

    @Override
    public void afterPropertiesSet() throws Exception {
        DefaultMQPushConsumer consumerOnline = initRocketMQConsumer("otp_order_yto_online_consumer", nameServer, onlineTopic);
        consumerOnline.setInstanceName(UUID.randomUUID().toString());
        consumerOnline.setMessageListener(new MessageListenerConcurrently(){
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : msgs) {
                    try{
                        orderFilterService.filterYTOOrder(new String(messageExt.getBody()), messageExt, true);
                    }catch (Exception e){
                        log.warn("consume message failed. messageExt:{}", messageExt, e);
                        context.setDelayLevelWhenNextConsume(0);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumerOnline.start();

        DefaultMQPushConsumer consumerOffline = initRocketMQConsumer("otp_order_yto_offline_consumer", nameServer, offlineTopic);
        consumerOffline.setInstanceName(UUID.randomUUID().toString());
        consumerOffline.setMessageListener(new MessageListenerConcurrently(){
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : msgs) {
                    try{
                        orderFilterService.filterYTOOrder(new String(messageExt.getBody()), messageExt, false);
                    }catch (Exception e){
                        log.warn("consume message failed. messageExt:{}", messageExt, e);
                        context.setDelayLevelWhenNextConsume(0);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumerOffline.start();
    }

    public DefaultMQPushConsumer initRocketMQConsumer(String group, String nameServer, String topic)throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeThreadMax(64);
        consumer.setConsumeThreadMin(20);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setVipChannelEnabled(true);
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.setPullBatchSize(32);
        consumer.setPollNameServerInterval(1000 * 30);
        consumer.setHeartbeatBrokerInterval(1000 * 30);
        consumer.subscribe(topic, "*");
        return consumer;
    }

}
